package easycall

import (
	"bytes"
	"fmt"
	"reflect"
	"runtime"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	errors "wanyu/server/engine/wperror"
	log "wanyu/server/engine/wplog"
)

//增大协程的处理能力
const DefaultAsyncCallCnt int = 10240

type CallFun struct {
	fun chan func()
	//timers map[int]*time.Timer
	times sync.Map
	//记录定时器id的最大值，假如每秒分配一个定时器，2的31次方可以用60多年，足够
	maxTimerId int32
	isClose    bool
	//为了防止本协程向本协程投递同步任务导致的死循环，需要做协程id判断（此判断很影响性能，仅在调试期使用）
	goroutingId uint64
}

func Create(funCnt int) *CallFun {
	re := &CallFun{
		fun: make(chan func(), funCnt),
		//timers:  make(map[int]*time.Timer),
		isClose: false,
	}
	re.start()
	//re.Post(re.print)
	return re
}

// 多线程调试、拍错用，后期会砍掉
func (ac *CallFun) GetGOID() uint64 {
	return ac.goroutingId
}

func (ac *CallFun) print() {
	var fn func()
	fn = func() {
		//fmt.Println("lens = ", len(ac.timers))
		ac.afterFunImpl(time.Second, fn)
	}
	ac.afterFunImpl(time.Second, fn)
}

func getFuncName(i func()) string {
	fn := runtime.FuncForPC(reflect.ValueOf(i).Pointer()).Name()
	return fmt.Sprintf("%v", fn)
}

func getFuncNameE(i func() error) string {
	fn := runtime.FuncForPC(reflect.ValueOf(i).Pointer()).Name()
	return fmt.Sprintf("%v", fn)
}

//提交一个函数来执行
func (ac *CallFun) Post(fun func()) {
	if ac.isClose {
		return
	}
	flag := false
	if len(ac.fun) >= DefaultAsyncCallCnt {
		// buf := make([]byte, 2048)
		// length := runtime.Stack(buf, false)
		flag = true
		log.Error("post channel is full, func name = ", getFuncName(fun))
	}
	ac.fun <- fun
	if flag {
		log.Error("post is success, fun name = ", getFuncName(fun))
	}
}

//同步调用
func (ac *CallFun) Call(fun func()) {
	//检测是否有本协程向本协程投递同步任务的情况
	errors.CheckError(getGoroutineID() != ac.goroutingId, fmt.Errorf("can not send function in same goroutine"))

	if ac.isClose {
		return
	}
	var wg sync.WaitGroup
	wg.Add(1)
	flag := false
	if len(ac.fun) >= DefaultAsyncCallCnt {
		// buf := make([]byte, 2048)
		// length := runtime.Stack(buf, false)
		flag = true
		log.Error("call channel is full, func name = ", getFuncName(fun))
	}
	ac.fun <- func() {
		ac.callWraper(fun)
		wg.Done()
	}
	wg.Wait()

	if flag {
		log.Error("call is success, fun name = ", getFuncName(fun))
	}
}

func (ac *CallFun) CallE(fun func() error) error {
	//检测是否有本协程向本协程投递同步任务的情况
	errors.CheckError(getGoroutineID() != ac.goroutingId, fmt.Errorf("can not send function in same goroutine"))

	if ac.isClose {
		return nil
	}
	var err error
	var wg sync.WaitGroup
	wg.Add(1)
	flag := false
	if len(ac.fun) >= DefaultAsyncCallCnt {
		// buf := make([]byte, 2048)
		// length := runtime.Stack(buf, false)
		// log.Error("callE channel is full, stack = ", string(buf[:length]))
		flag = true
		log.Error("callE channel is full, func name = ", getFuncNameE(fun))
	}
	ac.fun <- func() {
		err = ac.callWraperE(fun)
		wg.Done()
	}
	wg.Wait()
	if flag {
		log.Error("callE is success, fun name = ", getFuncNameE(fun))
	}
	return err
}

//延迟执行某个函数，返回定时器id
func (ac *CallFun) AfterFun(d time.Duration, fun func()) int {
	// ac.maxTimerId++
	// id := ac.maxTimerId
	id := atomic.AddInt32(&ac.maxTimerId, 1)
	//log.Info("once timer, -> ", id)
	wraper := func() {
		fun()
		ac.times.Delete(id)
		//delete(ac.timers, id)
	}
	//ac.timers[id] = ac.afterFunImpl(d, wraper)
	ac.times.Store(id, ac.afterFunImpl(d, wraper))

	return int(id)
}

func (ac *CallFun) afterFunImpl(d time.Duration, fun func()) *time.Timer {
	return time.AfterFunc(d, func() {
		ac.Post(fun)
	})
}

func (ac *CallFun) AddTimer(frequency time.Duration, fun func()) int {
	// ac.maxTimerId++
	// id := ac.maxTimerId
	id := atomic.AddInt32(&ac.maxTimerId, 1)

	//log.Info("add timer, -> ", id)
	var wraper func()
	wraper = func() {
		fun()
		//ac.timers[id] = ac.afterFunImpl(frequency, wraper)
		ac.times.Store(id, ac.afterFunImpl(frequency, wraper))
	}

	//ac.timers[id] = ac.afterFunImpl(frequency, wraper)
	ac.times.Store(id, ac.afterFunImpl(frequency, wraper))

	return int(id)
}

//必须用异步删除
func (ac *CallFun) RemoveTimer(id int) {
	//log.Info("remove timer, -> ", id)
	// if id <= 0 {
	// 	fmt.Println("cccc")
	// }
	if _, ok := ac.times.Load(int32(id)); !ok {
		//log.Info("remove not exsit timer, -> ", id)
		return
	}
	fn := func() {
		if value, ok2 := ac.times.Load(int32(id)); ok2 {
			timer := value.(*time.Timer)
			timer.Stop()
			//delete(ac.timers, id)
			ac.times.Delete(int32(id))
			//log.Info("remove timer, -> ", id, suc, len(ac.timers), ac.timers)
		}
	}
	ac.Post(fn)
}

//必须用异步删除
func (ac *CallFun) RemoveAllTimer() {
	//log.Info("remove all timer")
	fn := func() {

		rangeFn := func(k interface{}, v interface{}) bool {
			timer := v.(*time.Timer)
			timer.Stop()
			ac.times.Delete(k)
			return true
		}

		ac.times.Range(rangeFn)

		// for _, v := range ac.timers {
		// 	v.Stop()
		// }

		//ac.timers = make(map[int]*time.Timer)
		//ac.maxTimerId = 0
	}
	ac.Post(fn)
}

//同步移除定时器，不能自己协程撤销自己协程里运行的定时器
func (ac *CallFun) SyncRemoveAllTimer() {
	rangeFn := func(k interface{}, v interface{}) bool {
		timer := v.(*time.Timer)
		if !timer.Stop() {
			buf := make([]byte, 2048)
			length := runtime.Stack(buf, false)
			log.Warn("cannel timer is failed, ", string(buf[:length]))
		}
		ac.times.Delete(k)
		return true
	}

	ac.times.Range(rangeFn)
}

func (ac *CallFun) start() {
	go func() {
		ac.goroutingId = getGoroutineID()
		for {
			select {
			case fun, ok := <-ac.fun:
				if !ok {
					return
				}
				ac.callWraper(fun)
			}
		}
	}()
}

func (ac *CallFun) Stop() {
	if ac.isClose {
		return
	}
	ac.isClose = true
	close(ac.fun)
}

func (ac *CallFun) callWraper(fun func()) {
	defer errors.Catch(nil)
	fun()
}

func (ac *CallFun) callWraperE(fun func() error) (err error) {
	defer errors.Catch(&err)
	return fun()
}

//获取协程是否关闭
func (ac *CallFun) IsClose() bool {
	return ac.isClose
}

func getGoroutineID() uint64 {
	b := make([]byte, 64)
	runtime.Stack(b, false)
	b = bytes.TrimPrefix(b, []byte("goroutine "))
	b = b[:bytes.IndexByte(b, ' ')]
	n, _ := strconv.ParseUint(string(b), 10, 64)
	return n
}
